package kafka.examples;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

public class ProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();

        // 这里可以配置几台broker即可，他会自动从broker去拉取元数据进行缓存
        props.put("bootstrap.servers", "pytha3:9092,pytha4:9092,pytha5:9092");
        // 这个就是负责把发送的key从字符串序列化为字节数组
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // 这个就是负责把你发送的实际的message从字符串序列化为字节数组
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //默认是none，不压缩，也可使用lz4压缩，压缩之后可以减小数据量，提升吞吐量，但是会加大producer端的cpu开销。
        props.put("compression.type", "lz4");

        //发送消息的缓冲区，默认值32MB，如果发送消息出去的速度小于写入消息进去的速度，就会导致缓冲区写满，此时生产消息就会阻塞住
        props.put("buffer.memory", 33554432);
        //设置每个bitch的大小，默认16KB，16384, 建议64K，128K
        props.put("batch.size", 323840);
        //设置每个bitch的发送间隔时间，默认是0，建议100ms，比凑成1个batch的时间稍长一点。
        props.put("linger.ms", 10);

        //设置发送1条消息的最大值，默认 1048576，1MB，太小，建议10MB。
        props.put("max.request.size",10485760);
        //该参数指定了阻塞时间，当生产者的缓冲区已满，这些方法就会阻塞。在阻塞时间达到max.block.ms时，生产者抛出超时异常。
        props.put("max.block.ms", 3000);

        /**
         * acks = 1; 只要leader partition写入成功，就认为消息写入成功。数据会丢失。leader刚接收到消息就挂了。 一般用这个。
         * acks = 0;  producer根本不管写入broker的消息到底成功没有，发送一条消息出去，立马发送吓一跳消息。吞吐量最高，会丢消息，适用于实时数据流分析，丢几条消息影响不大。
         * acks = -1; 等同于 acks=all; leader partition写入成功之后，必须等到其他ISR的副本都写入成功，才可以返回相应这条消息写入成功，会受到消息回调通知。消息不会丢失。
         *     min.insync.replicas=2; ISR里面必须有2个副本。
         *     retries= Integer.MAX_VALUE; 如果上面2个条件不满足，写入一直失败，就会无限重试。金融级场景，和钱相关的业务。
         */
        props.put("acks", "-1");

        /**
         * 一般设置3 ~5次即可满足大部分异常场景。 broker宕机导致leader切换时最常见的异常。
         *     会导致消息重复消费和，如网络波动导致误认为消息没有发送成功。消费者需要做幂等性处理。
         *     会导致消息乱序。max.in.flight.requests.per.connection=1,保证同一时间只能发送一条消息。
         */
        props.put("retries", 3);
        //两次重试之间的间隔。默认100ms/
        props.put("retry.backoff.ms", 100);

        // 创建一个Producer实例：线程资源，跟各个broker建立socket连接资源
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "test-key", "test-value");

        long startTime = System.currentTimeMillis();
        // 这是异步发送的模式
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if(exception == null) {
                    // 消息发送成功
                    System.out.println("消息发送成功");
                } else {
                    // 消息发送失败，需要重新发送
                }
            }
        });


        long endTime = System.currentTimeMillis();
        if(endTime-startTime > 10){
            //如果超过10秒，记录日志，向管理员发送报错信息
            //应该走一个监控和报警的过程，开发一些应用程序，系统
            //metric监控和报警，小米的open Falcon 隼(sun)监控
            //立马给程序员发短信，或者发送钉钉，或者发邮件
        }

        Thread.sleep(10 * 1000);

        // 这是同步发送的模式
//		producer.send(record).get();
        // 你要一直等待人家后续一系列的步骤都做完，发送消息之后
        // 有了消息的回应返回给你，你这个方法才会退出来

        producer.close();
    }
}
